create the kafka foundations#1
Conversation
📝 WalkthroughWalkthroughThis PR establishes a complete Spring Boot Kafka demo application with Maven build configuration, Docker Compose infrastructure, producer and consumer services, and a REST API endpoint. The application accepts messages via HTTP POST and publishes them to a Kafka topic that is consumed and logged. ChangesSpring Boot Kafka Demo Application
Sequence DiagramsequenceDiagram
participant Client as HTTP Client
participant API as MessageController
participant Producer as KafkaProducer
participant Kafka as Kafka Broker
participant Consumer as KafkaConsumer
Client->>API: POST /api/1/messages (message body)
API->>Producer: sendMessage(message)
Producer->>Kafka: publish to test-topic
Kafka->>Consumer: deliver message
Consumer->>Consumer: log message
API->>Client: 200 OK "Message queued successfully"
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In @.mvn/wrapper/maven-wrapper.properties:
- Around line 1-3: Add a distributionSha512Sum property to the Maven wrapper
properties to verify the downloaded archive integrity: update the
.mvn/wrapper/maven-wrapper.properties by adding the
distributionSha512Sum=<official SHA-512 checksum for
apache-maven-3.9.15-bin.zip> alongside the existing wrapperVersion,
distributionType, and distributionUrl entries so the wrapper will validate the
downloaded archive using the official SHA-512 checksum.
In `@src/main/java/com/alsainey/kafka_demo/config/KafkaTopicConfig.java`:
- Around line 4-9: The class KafkaTopicConfig is annotated with `@Configurable`
(AspectJ weaving) so its `@Bean` methods won't be registered; replace
`@Configurable` with Spring's `@Configuration` to ensure the `@Bean` factory methods
are discovered and loaded into the application context (i.e., change the class
annotation from `@Configurable` to `@Configuration` so the `@Bean` methods in
KafkaTopicConfig are properly registered).
In `@src/main/java/com/alsainey/kafka_demo/consumer/KafkaConsumer.java`:
- Line 19: The log call in KafkaConsumer that does log.info(format("Message
consumed: %s", message)) exposes full message payloads; change it to avoid
logging raw message content at INFO level by logging only non-sensitive metadata
(e.g., topic, partition, offset, key) or by redacting/sanitizing the message,
and if payload inspection is needed keep the full message at DEBUG level (e.g.,
use log.debug for the full message) so replace the current log.info usage
accordingly referencing the KafkaConsumer class, the log variable and the
message variable.
In `@src/main/java/com/alsainey/kafka_demo/producer/KafkaProducer.java`:
- Around line 18-21: Change sendMessage to return the future produced by
kafkaTemplate.send so callers can observe success/failure: update the signature
of sendMessage(...) to return the producer future type returned by
kafkaTemplate.send(...) (e.g., CompletableFuture/
ListenableFuture<SendResult<...>> depending on your KafkaTemplate generics) and
return kafkaTemplate.send("test-topic", message) instead of discarding it; then
update MessageController to consume that future and map success to 200 OK and
failures to an appropriate error response so send result failures propagate to
the HTTP client. Ensure you reference the existing sendMessage method and
kafkaTemplate.send call and adapt generics to match the KafkaTemplate used.
In `@src/main/java/com/alsainey/kafka_demo/rest/MessageController.java`:
- Around line 20-24: In MessageController (the REST handler that accepts
`@RequestBody` String message and calls kafkaProducer.sendMessage), add input
validation to reject null or blank messages: check message for null or
message.trim().isEmpty() before calling kafkaProducer.sendMessage and if blank
return ResponseEntity.status(HttpStatus.BAD_REQUEST).body(...) (400) with an
explanatory message; otherwise proceed to call
kafkaProducer.sendMessage(message) and return the existing 200 response.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: b372cbbf-1593-49b2-b28f-993b63e94d47
📒 Files selected for processing (14)
.gitattributes.gitignore.mvn/wrapper/maven-wrapper.propertiesdocker-compose.ymlmvnwmvnw.cmdpom.xmlsrc/main/java/com/alsainey/kafka_demo/KafkaDemoApplication.javasrc/main/java/com/alsainey/kafka_demo/config/KafkaTopicConfig.javasrc/main/java/com/alsainey/kafka_demo/consumer/KafkaConsumer.javasrc/main/java/com/alsainey/kafka_demo/producer/KafkaProducer.javasrc/main/java/com/alsainey/kafka_demo/rest/MessageController.javasrc/main/resources/application.yamlsrc/test/java/com/alsainey/kafka_demo/KafkaDemoApplicationTests.java
| wrapperVersion=3.3.4 | ||
| distributionType=only-script | ||
| distributionUrl=https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.9.15/apache-maven-3.9.15-bin.zip |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
url="$(awk -F= '$1=="distributionUrl"{print $2}' .mvn/wrapper/maven-wrapper.properties)"
echo "distributionUrl=$url"
echo "Official SHA256 (from repository sidecar):"
curl -fsSL "${url}.sha256"
echo
echo "Configured checksum line (should exist):"
grep -n '^distributionSha256Sum=' .mvn/wrapper/maven-wrapper.properties || trueRepository: Juniorbarry26/kafka
Length of output: 292
🏁 Script executed:
# Try alternative methods to find official SHA256 for Maven 3.9.15
echo "=== Checking Maven Central directly ==="
# Try the base URL without .sha256
curl -fsSL -I "https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.9.15/apache-maven-3.9.15-bin.zip"
echo -e "\n=== Checking for .asc (GPG signature) file ==="
curl -fsSL -I "https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.9.15/apache-maven-3.9.15-bin.zip.asc"
echo -e "\n=== Checking for checksums.txt ==="
curl -fsSL -I "https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.9.15/checksums.txt"
echo -e "\n=== Checking Apache Archives for checksum ==="
curl -fsSL -I "https://archive.apache.org/dist/maven/maven-3/3.9.15/binaries/apache-maven-3.9.15-bin.zip.sha512"
echo -e "\n=== Listing what's available in Maven Central for 3.9.15 ==="
curl -fsSL "https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.9.15/" 2>/dev/null | grep -oP 'href="[^"]*"' | head -20 || trueRepository: Juniorbarry26/kafka
Length of output: 2682
🏁 Script executed:
echo "=== Fetching official SHA512 from Apache Archives ==="
curl -fsSL "https://archive.apache.org/dist/maven/maven-3/3.9.15/binaries/apache-maven-3.9.15-bin.zip.sha512"
echo -e "\n=== Fetching SHA512 from Maven Central ==="
curl -fsSL "https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.9.15/apache-maven-3.9.15-bin.zip.sha512"
echo -e "\n=== Checking Maven Wrapper documentation and source ==="
# Look for Maven wrapper configuration documentation
find . -name "*wrapper*" -o -name "*.md" | grep -i readme || true
ls -la .mvn/ 2>/dev/null || trueRepository: Juniorbarry26/kafka
Length of output: 651
🏁 Script executed:
echo "=== Checking Maven wrapper files ==="
ls -la .mvn/wrapper/
echo -e "\n=== Current maven-wrapper.properties content ==="
cat .mvn/wrapper/maven-wrapper.properties
echo -e "\n=== Searching for Maven wrapper documentation ==="
find . -type f \( -name "*.md" -o -name "*.txt" -o -name "*.xml" \) -exec grep -l "distributionSha256Sum\|distributionSha512Sum" {} \; 2>/dev/null || echo "No matches found"
echo -e "\n=== Checking pom.xml or maven-wrapper plugin config ==="
grep -A 20 -B 5 "maven-wrapper\|distributionSha" ./pom.xml 2>/dev/null || echo "Not found in pom.xml"Repository: Juniorbarry26/kafka
Length of output: 629
🌐 Web query:
Maven wrapper properties distributionSha256Sum distributionSha512Sum supported
💡 Result:
The Maven Wrapper supports both SHA-256 and SHA-512 checksum verification for the Maven distribution and the Maven Wrapper JAR [1][2][3]. The following properties can be configured in the.mvn/wrapper/maven-wrapper.properties file: - distributionSha256Sum: The expected SHA-256 checksum of the Maven distribution [4][2]. - wrapperSha256Sum: The expected SHA-256 checksum of the maven-wrapper.jar [4][2]. - distributionSha512Sum: The expected SHA-512 checksum of the Maven distribution [1][3]. - wrapperSha512Sum: The expected SHA-512 checksum of the maven-wrapper.jar [1][3]. Support for SHA-512 checksums was introduced to align with the checksum formats published by the Apache Maven project, as Maven distributions are primarily published with SHA-512 checksums rather than SHA-256 [1][5]. While SHA-256 remains supported for backward compatibility, SHA-512 is the recommended approach for verifying modern Maven distributions [1]. When using these properties, the Maven Wrapper will validate the downloaded artifact against the provided checksum before execution [2][6]. If the checksum does not match, the wrapper will fail to prevent the execution of potentially compromised or corrupted files [1][6]. Note that checksum validation is not supported for the Maven Daemon (mvnd) [3].
Citations:
- 1: [MWRAPPER-117] added distributionSha512Sum and wrapperSha512Sum apache/maven-wrapper#398
- 2: https://maven.apache.org/tools/wrapper/
- 3: https://mail-archive.com/issues@maven.apache.org/msg341667.html
- 4: https://maven.apache.org/tools/wrapper/maven-wrapper-plugin/wrapper-mojo.html
- 5: [MWRAPPER-117] Wrapper distributions and jars sha256 checksums should be published apache/maven-wrapper#316
- 6: https://relentlesscoding.com/posts/maven-wrapper-integrity-validation/
Add distributionSha512Sum to verify Maven distribution integrity.
Line 3 downloads an executable archive, but checksum verification is not configured. This leaves the wrapper without content-integrity pinning. Apache Maven officially publishes SHA-512 checksums (not SHA-256); configure verification using the official checksum:
wrapperVersion=3.3.4
distributionType=only-script
distributionUrl=https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.9.15/apache-maven-3.9.15-bin.zip
+distributionSha512Sum=251c676eb88684d6dd86c666817979d256a335c31ad34ab0e1df17eeca1cd504a6c890e906a46d04e51cc7a4fb02f33a75971ef750558034e0a81c257be02267📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| wrapperVersion=3.3.4 | |
| distributionType=only-script | |
| distributionUrl=https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.9.15/apache-maven-3.9.15-bin.zip | |
| wrapperVersion=3.3.4 | |
| distributionType=only-script | |
| distributionUrl=https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.9.15/apache-maven-3.9.15-bin.zip | |
| distributionSha512Sum=251c676eb88684d6dd86c666817979d256a335c31ad34ab0e1df17eeca1cd504a6c890e906a46d04e51cc7a4fb02f33a75971ef750558034e0a81c257be02267 |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In @.mvn/wrapper/maven-wrapper.properties around lines 1 - 3, Add a
distributionSha512Sum property to the Maven wrapper properties to verify the
downloaded archive integrity: update the .mvn/wrapper/maven-wrapper.properties
by adding the distributionSha512Sum=<official SHA-512 checksum for
apache-maven-3.9.15-bin.zip> alongside the existing wrapperVersion,
distributionType, and distributionUrl entries so the wrapper will validate the
downloaded archive using the official SHA-512 checksum.
|
|
||
| @KafkaListener(topics = "test-topic", groupId = "test-group") | ||
| public void consumeMessage(String message) { | ||
| log.info(format("Message consumed: %s", message)); |
There was a problem hiding this comment.
Avoid logging raw message payloads at INFO level.
Line 19 logs the full message body, which can leak PII/secrets into logs.
✅ Proposed fix
- log.info(format("Message consumed: %s", message));
+ log.info("Message consumed (length={})", message == null ? 0 : message.length());📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| log.info(format("Message consumed: %s", message)); | |
| log.info("Message consumed (length={})", message == null ? 0 : message.length()); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/main/java/com/alsainey/kafka_demo/consumer/KafkaConsumer.java` at line
19, The log call in KafkaConsumer that does log.info(format("Message consumed:
%s", message)) exposes full message payloads; change it to avoid logging raw
message content at INFO level by logging only non-sensitive metadata (e.g.,
topic, partition, offset, key) or by redacting/sanitizing the message, and if
payload inspection is needed keep the full message at DEBUG level (e.g., use
log.debug for the full message) so replace the current log.info usage
accordingly referencing the KafkaConsumer class, the log variable and the
message variable.
| public void sendMessage(String message) { | ||
| log.info(format("Sending message: %s", message)); | ||
| kafkaTemplate.send("test-topic", message); | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🌐 Web query:
For the Spring Kafka version used by this project, what is the exact return type of KafkaTemplate.send(...), and what is the recommended way to handle publish failures?
💡 Result:
For the Spring Kafka version used by your project (Spring Kafka’s KafkaTemplate API): 1) Exact return type of KafkaTemplate.send(...) KafkaTemplate.send(...) returns a CompletableFuture of SendResult: CompletableFuture<SendResult<K, V>> [1][2] 2) Recommended way to handle publish failures Handle failures asynchronously by attaching a completion callback to the returned future (e.g., whenComplete / callback), and react when the future completes exceptionally (or when you call get(...) with a timeout). Spring Kafka’s documentation shows registering a callback on the returned CompletableFuture to receive either the SendResult or an exception. [2][3] If you need strict “only return success to the client when the broker acknowledged the send”, the docs also recommend blocking only by waiting on the future, ideally with a timeout (future.get(...) / with timeout) and handling ExecutionException (send failure) separately. [3] Caveat: there are producer-side failures that can be thrown immediately, and others delivered via the future completing exceptionally; using the future result/callback is the recommended approach because it captures both outcomes. [4][2]
Citations:
- 1: https://docs.spring.io/spring-kafka/docs/current-SNAPSHOT/api/org/springframework/kafka/core/KafkaTemplate.html
- 2: https://docs.enterprise.spring.io/spring-kafka/reference/kafka/sending-messages.html
- 3: https://docs.spring.io/spring-kafka/reference/kafka/sending-messages.html
- 4: https://github.com/spring-projects/spring-kafka/blob/master/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java
🏁 Script executed:
# Find build configuration files to determine Spring Kafka version
fd -e pom.xml -e build.gradle -e gradle.properties -e "*.gradle.kts" | head -5Repository: Juniorbarry26/kafka
Length of output: 45
🏁 Script executed:
# Check the exact content at the specified lines in the file
cat -n src/main/java/com/alsainey/kafka_demo/producer/KafkaProducer.java | head -30Repository: Juniorbarry26/kafka
Length of output: 773
🏁 Script executed:
# Search for all kafkaTemplate.send calls in the codebase to understand the pattern
rg "kafkaTemplate\.send" --type java -nRepository: Juniorbarry26/kafka
Length of output: 183
🏁 Script executed:
# Search more broadly for build configuration files
find . -type f \( -name "pom.xml" -o -name "build.gradle" -o -name "build.gradle.kts" \) 2>/dev/nullRepository: Juniorbarry26/kafka
Length of output: 73
🏁 Script executed:
# Also check parent directories or alternative locations
ls -la | grep -E "build|gradle|pom|maven"Repository: Juniorbarry26/kafka
Length of output: 121
🏁 Script executed:
# Read the pom.xml to find Spring Kafka version
cat pom.xmlRepository: Juniorbarry26/kafka
Length of output: 2976
🏁 Script executed:
# Search for all calls to sendMessage method
rg "sendMessage" --type java -nRepository: Juniorbarry26/kafka
Length of output: 407
🏁 Script executed:
# Check if there's a controller that calls this producer
fd -e java -path "*/controller/*" | head -10Repository: Juniorbarry26/kafka
Length of output: 232
🏁 Script executed:
# Read the MessageController to understand the full context
cat -n src/main/java/com/alsainey/kafka_demo/rest/MessageController.javaRepository: Juniorbarry26/kafka
Length of output: 1061
Return the send result so failures can be propagated to the caller.
Line 20 discards the CompletableFuture<SendResult<K, V>> returned by kafkaTemplate.send(), making it impossible to detect or handle publish failures. The HTTP response at line 24 of MessageController returns 200 OK regardless of whether the Kafka send succeeds or fails, giving clients false confidence that the message was enqueued.
Modify sendMessage() to return the future and let the controller map the result to the HTTP response status. This way, clients can distinguish between successful enqueue and failures.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/main/java/com/alsainey/kafka_demo/producer/KafkaProducer.java` around
lines 18 - 21, Change sendMessage to return the future produced by
kafkaTemplate.send so callers can observe success/failure: update the signature
of sendMessage(...) to return the producer future type returned by
kafkaTemplate.send(...) (e.g., CompletableFuture/
ListenableFuture<SendResult<...>> depending on your KafkaTemplate generics) and
return kafkaTemplate.send("test-topic", message) instead of discarding it; then
update MessageController to consume that future and map success to 200 OK and
failures to an appropriate error response so send result failures propagate to
the HTTP client. Ensure you reference the existing sendMessage method and
kafkaTemplate.send call and adapt generics to match the KafkaTemplate used.
| @RequestBody String message | ||
| ) { | ||
| kafkaProducer.sendMessage(message); | ||
|
|
||
| return ResponseEntity.ok("Message queued successfully"); |
There was a problem hiding this comment.
Reject blank messages before publishing.
Line 20–24 forwards request body without validation; blank payloads should return 400 Bad Request.
✅ Proposed fix
public ResponseEntity<String> sendMessage(
`@RequestBody` String message
) {
+ if (message == null || message.isBlank()) {
+ return ResponseEntity.badRequest().body("Message must not be blank");
+ }
kafkaProducer.sendMessage(message);
return ResponseEntity.ok("Message queued successfully");
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| @RequestBody String message | |
| ) { | |
| kafkaProducer.sendMessage(message); | |
| return ResponseEntity.ok("Message queued successfully"); | |
| public ResponseEntity<String> sendMessage( | |
| `@RequestBody` String message | |
| ) { | |
| if (message == null || message.isBlank()) { | |
| return ResponseEntity.badRequest().body("Message must not be blank"); | |
| } | |
| kafkaProducer.sendMessage(message); | |
| return ResponseEntity.ok("Message queued successfully"); | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/main/java/com/alsainey/kafka_demo/rest/MessageController.java` around
lines 20 - 24, In MessageController (the REST handler that accepts `@RequestBody`
String message and calls kafkaProducer.sendMessage), add input validation to
reject null or blank messages: check message for null or
message.trim().isEmpty() before calling kafkaProducer.sendMessage and if blank
return ResponseEntity.status(HttpStatus.BAD_REQUEST).body(...) (400) with an
explanatory message; otherwise proceed to call
kafkaProducer.sendMessage(message) and return the existing 200 response.
1 participant
Summary by CodeRabbit
New Features
Chores
Tests